package com.dhcy.mqtt.service.impl;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dhcy.mqtt.config.mqtt.MqttConfig;
import com.dhcy.mqtt.model.OnlineData;
import com.dhcy.mqtt.par.ht.RevEquipmentDataAddPar;
import com.dhcy.mqtt.service.OnlineDataService;
import com.dhcy.mqtt.service.RevEquipmentDataService;
import com.dhcy.mqtt.util.OkhttpUtil;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Service;

import java.util.Date;


@Service
@Slf4j
public class MqttServiceImpl {
    @Autowired
    MessageChannel inChannel;

    //@Autowired
    //MessageChannel outChannel;

    @Autowired
    MqttConfig mqttConfig;

    @Autowired
    MqttPahoClientFactory mqttPahoClient;

    @Autowired
    OkhttpUtil okhttpUtil;

    @Autowired
    OnlineDataService onlineDataService;


    // @Bean
    //@ServiceActivator(inputChannel = "outChannel")
    //public MessageHandler mqttMsgSend() {
    //    MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getClientId() + "11", mqttPahoClient);
    //    messageHandler.setAsync(true);
    //    return messageHandler;
    //}


    //以下为mqtt消息接收逻辑
    // MQTT 消息订阅绑定（消费者）
    @Bean
    public MessageProducer mqttMsgRecv() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                mqttConfig.getClientId(), mqttPahoClient, mqttConfig.getTopic().split(","));
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        //设置消息质量：0->至多一次；1->至少一次；2->只有一次
        adapter.setQos(0);
        adapter.setOutputChannel(inChannel);
        return adapter;
    }

    // MQTT消息处理器（消费者，用于服务端自发自收的测试）
    @Bean
    @ServiceActivator(inputChannel = "inChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String receivemessage = message.getPayload().toString();
                log.info("handleMessage:{}", message.getPayload());
                onlineDataService.add(receivemessage);
            }
        };
    }
}
